package com.wz.jiangsu.wangteng.kafka.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class kafkaProducer {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;



    @Transactional
    @GetMapping("/kafka/normal/{message}")
    public String sendNormalMessage(@PathVariable("message") String message) {
        String string = kafkaTemplate.toString();
        System.out.println(string);
        ListenableFuture<SendResult<String, Object>> sbTopic = kafkaTemplate.send("test_zzx", message);

        final String[] msg = new String[1];
        sbTopic.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                if (ex != null) {
                    msg[0] = "ex.getMessage() = " + ex.getMessage();
                    System.out.println(msg[0]);
                }
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                msg[0] = "wangtengtech" + result.getRecordMetadata().topic();
                System.out.println(msg[0]);
            }
        });
        // 先走完方法再发送消息后面再回调，导致 msg[0] 没有赋值进去
        return message;
    }

    @GetMapping("/kafka/transaction/{message}")
    public void sendTransactionMessage(@PathVariable("message") String message) {
        //声明事务：后面报错消息不会发出去
        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<String, Object, Object>() {

            @Override
            public Object doInOperations(KafkaOperations<String, Object> operations) {
                operations.send("sb_topic", message + " test executeInTransaction");
                throw new RuntimeException("fail");
            }
        });
        // //不声明事务：后面报错但前面消息已经发送成功了
        // kafkaTemplate.send("sb_topic", message + " test executeInNoTransaction");
        // throw new RuntimeException("fail");
    }

}
